package com.shujia.city.dwd

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object DwdRsZhcsAz01bsbkxxMskDay extends SparkTool {
  /**
   * 在run函数中写数据处理的逻辑
   * import spark.implicits._
   * import org.apache.spark.sql.functions._
   */
  override def run(spark: SparkSession, ds: String): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._
    //读取数据
    val az02bsbkxx: Dataset[Row] = spark
      .table("ods.ods_t_rs_zhcs_az01bsbkxx")
      .where($"ds" === ds)

    //处理数据
    val resultDF: DataFrame = az02bsbkxx
      .select(
        upper(md5(concat($"id", expr("'shujia'")))) as "id",
        $"citty_id",
        $"ss_id",
        $"fkrq",
        $"yxqz",
        $"aaz502",
        regexp_replace($"aae008", "\"", "") as "aae008",
        $"aae008b",
        $"aae010",
        regexp_replace($"aae010a", "\"", "") as "aae010a",
        $"aae010b"
      )

    //保存数据
    saveToTable(spark, resultDF, "dwd.dwd_rs_zhcs_az01bsbkxx_msk_d", ds)
  }
}
